package net.tw.transactional2.task.impl;

import lombok.extern.slf4j.Slf4j;
import net.tw.transactional2.task.TaskBService;
import net.tw.transactional2.task.TaskCService;
import org.aspectj.lang.ProceedingJoinPoint;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.ResponseEntity;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.util.Hashtable;

@Service
@Slf4j
@SuppressWarnings("unchecked")
public class TaskBServiceImpl implements TaskBService {

    @Autowired
    DataSourceTransactionManager dataSourceTransactionManager;

    @Autowired
    DefaultTransactionDefinition transactionDefinition;

    @Autowired
    @Qualifier("threadExchangeTable")
    Hashtable<Long, Object> threadExchangeTable;

    @Value("${spring.application.name}")
    String appName;

    @Autowired
    TaskCService taskCService;

    @Async("taskExecutor")
    @Override
    public void taskB(Long threadId, ProceedingJoinPoint jp) {

        //------------------------业务处理---------------------------

        Thread currentThread = Thread.currentThread();
        Long currentThreadId = currentThread.getId();
        String currentThreadName = currentThread.getName();
        TransactionStatus transactionStatus = null;
        ResponseEntity<Object> retObj = null;

        Hashtable<String, Object> oneThreadExchangeData = new Hashtable<>();
        oneThreadExchangeData.put("thread_id", String.valueOf(currentThreadId));
        oneThreadExchangeData.put("thread_name", currentThreadName);
        oneThreadExchangeData.put("app_name", appName);

        try {
            transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
            retObj = (ResponseEntity<Object>) jp.proceed();

            oneThreadExchangeData.put("ret_value", retObj.getBody());
            oneThreadExchangeData.put("status", true);
            threadExchangeTable.put(threadId, oneThreadExchangeData);
            taskCService.syncTaskC(threadId);

        } catch (Throwable th) {
            oneThreadExchangeData.put("status", false);
            oneThreadExchangeData.put("ret_value", th.getMessage());
            threadExchangeTable.put(threadId, oneThreadExchangeData);
            taskCService.syncTaskC(threadId);
            return;
        }

        //---------------------事务提交-----------------------------

        Hashtable<String, Object> twoThreadExchangeData = new Hashtable<>();
        Long commitThreadId = -1l;
        String action = null;
        String commitMsg = null;

        synchronized (currentThread) {
            try {
                currentThread.wait();

                twoThreadExchangeData = (Hashtable<String, Object>) threadExchangeTable.get(currentThreadId);
                commitThreadId = Long.valueOf(String.valueOf(twoThreadExchangeData.get("thread_id")));
                action = twoThreadExchangeData.get("action").toString();

                if (action.equals("commit")) {
                    dataSourceTransactionManager.commit(transactionStatus);
                    commitMsg = "线程[" + currentThreadId + "]-事务完成-[确认]";
                } else {
                    dataSourceTransactionManager.rollback(transactionStatus);
                    commitMsg = "线程[" + currentThreadId + "]-事务完成-[回滚]";
                }
                log.info(commitMsg);
                threadExchangeTable.put(commitThreadId, commitMsg);
                taskCService.syncTaskC(commitThreadId);
            } catch (TransactionException e) {
                commitMsg = "线程[" + currentThreadId + "]-事务提交或回滚异常-[事务失败]";
                log.info(commitMsg);
                threadExchangeTable.put(commitThreadId, commitMsg);
                taskCService.syncTaskC(commitThreadId);
            } catch (InterruptedException e) {
                dataSourceTransactionManager.rollback(transactionStatus);
                commitMsg = "线程[" + currentThreadId + "]-异常中断-[事务回滚]";
                log.info(commitMsg);
            } finally {
                threadExchangeTable.remove(currentThreadId);
            }
        }
    }
}
